package com.lvmama.rhino.analyze.client

import com.lvmama.rhino.analyze.processing.WirelessV2Search
import com.lvmama.rhino.common.entity.{JobConfig, KafkaTopic, SEARCH, WirelessClientMonitorLog}
import com.lvmama.rhino.common.utils.JDBCUtil.Conversions
import com.lvmama.rhino.common.utils.Utils
import com.lvmama.rhino.common.utils.kafka.{KafkaDStreamSource, KafkaPayloadStringCodec}
import com.lvmama.rhino.common.utils.logging.LazyLogging
import com.lvmama.rhino.common.utils.sharedvariable.SessionAccumulator
import com.lvmama.rhino.common.utils.spark.{SparkStreamingApplication, StructuredStreamingApplication}
import org.apache.commons.codec.binary.Base64
import org.apache.spark.SparkContext
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.{Dataset, TypedColumn}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.{OutputMode, ProcessingTime}
import org.apache.spark.sql.types.{StringType, StructType}

import scala.collection.mutable.{ArrayBuffer, ListBuffer}

/**
  * Created by yuanxiaofeng on 2017/3/15.
  */
class WirelessStatV2(config: JobConfig) extends StructuredStreamingApplication with LazyLogging{

  override var sparkConfig: Map[String, String] = config.spark
  override var appName: String = "WirelessStatV2Job"
  override def streamingBatchDuration: Long = 0L
  override def streamingCheckpointDir: String = ???

  val kafkaConf = config.sourceKafka

  private val kafkaSourceTopic = KafkaTopic("wireless_v2")
  private val kafkaSinkTopic = KafkaTopic("WirelessStatV2Sink")

  @transient var sc: SparkContext = _

  def execute() = {
//    sparkConfig += ("spark.executor.memory" -> "3g")
    sparkConfig += ("spark.sql.autoBroadcastJoinThreshold" -> "10485760")
    withStructuredApplication{spark =>
      sc = spark.sparkContext
      val sessionAcc = WirelessStatV2.getInstanceSessionAcc
      sc.register(sessionAcc, "sessionAccumulator")

      import spark.implicits._
      val streaming = spark
        .readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", kafkaConf.get("kafka.bootstrap.servers").get)
        .option("subscribe", kafkaSourceTopic.name)
        .load()
        .selectExpr("CAST(value AS STRING)")
        .as[String]
        .map(l => Utils.initWireless(l))
        .as[WirelessClientMonitorLog]

      import Conversions._
      val predicates = Array("CodeTypeLevel = 'D2'")
      val deviceCode = spark
        .jdbc("statistic_continuity_code", predicates)
        .select(col("code").as("platformCode"), col("name"))

      val assignSessionId = udf((deviceToken: String, timeStamp: Long) => {
        sessionAcc.addWithResult(deviceToken, timeStamp)
      })


      val wireless = streaming.withColumn("sessionId", assignSessionId(col("deviceToken"), col("timeStamp")))
        .join(deviceCode, Seq("platformCode"), "left")
        .drop(col("platformCode"))
        .withColumnRenamed("name", "platformCode")
        .as[WirelessClientMonitorLog]

      new WirelessV2Search().process(wireless)
      import scala.concurrent.duration._
//      val input = wireless
//        .coalesce(1)
//        .write
//        .parquet("hdfs://10.200.2.104:8020/yxf/wireless")

//        .writeStream
//        .format("console")
//        .queryName("test")
//        .outputMode(OutputMode.Append())
//        .option("checkpointLocation", "hdfs://10.200.2.104:8020/checkpoint/wireless_v2")
//        .trigger(ProcessingTime(1.minutes))
//        .start()
//      input.awaitTermination()
    }
  }
}

object WirelessStatV2 {
  def apply() = {
    val config = JobConfig()
    new WirelessStatV2(config).execute()
  }

  @volatile private var _sessionAcc: SessionAccumulator = null

  def getInstanceSessionAcc: SessionAccumulator = {
    if (_sessionAcc == null) {
      synchronized {
        if (_sessionAcc == null) {
          _sessionAcc = new SessionAccumulator
        }
      }
    }
    _sessionAcc
  }

  @volatile private var _deviceBro: Broadcast[Map[Any, Any]] = null

  def getInstanceDeviceBro(sc:SparkContext, deviceCode: Map[Any, Any]): Broadcast[Map[Any, Any]] = {
    if (_deviceBro == null) {
      synchronized {
        if (_deviceBro == null) {
          _deviceBro = sc.broadcast(deviceCode)
        }
      }
    }
    _deviceBro
  }
}
